package cn.com.bluemoon.daps.sync.core.sync.executor;

import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.enums.SyncStatus;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.common.toolkit.SshUtils;
import cn.com.bluemoon.daps.common.toolkit.valueobject.PtSsh;
import cn.com.bluemoon.daps.sync.base.BmScript;
import cn.com.bluemoon.daps.sync.base.CompareResult;
import cn.com.bluemoon.daps.sync.base.ModifyType;
import cn.com.bluemoon.daps.sync.base.SqlCompareResult;
import cn.com.bluemoon.daps.sync.config.PtServerConfig;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.compare.TableCompareRes;
import cn.com.bluemoon.daps.sync.core.sync.log.EvenType;
import cn.com.bluemoon.daps.sync.core.sync.log.LogUtils;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvEventFactory;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvLog;
import cn.com.bluemoon.daps.sync.core.sync.ptosc.PtOnlineSchemaChange;
import cn.com.bluemoon.daps.sync.core.sync.sql.mysql.MysqlConnect;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.parser.ParserException;
import lombok.extern.slf4j.Slf4j;
import reactor.util.function.Tuple2;

import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * mysql ptosc 工具执行器
 * 注：
 * ptosc只支持执行非表级别的变更ddl
 * 比如rename table xx to xxx 就无法执行
 * 因此会在执行脚本中能有区分出那些是表级别的ddl,目前统一通过{@link CompareResult#getModifyType()}为table时候为表级别ddl
 * 此时，会用原生的jdbc{@link JdbcSqlExecutor#execute()}进行执行
 * 其他非表级别的会用ptosc执行
 *
 * @author Jarod.Kong
 * @date 2021-06-05 11:08
 **/
@Slf4j
public class PtSqlExecutor extends SqlExecutor {
    private final JdbcSqlExecutor jdbcExecutor;

    public PtSqlExecutor(GlobalConfig globalConfig) {
        super(ExecutorType.PT, globalConfig);
        this.jdbcExecutor = (JdbcSqlExecutor) SqlExecutors.jdbc(globalConfig);
        BmAssetUtils.isTrue(DatabaseType.MYSQL.equals(globalConfig.getDataSource().getDatabaseType()), "ptosc只支持mysql数据库");
    }

    @Override
    public boolean execute0() {
        // 执行表
        // 2021/6/28 增加收集每个表变更执行的结果情况，用于返回是否都是全部执行成功
        List<TableCompareRes<SqlCompareResult>> tableInfoSqlResult = getTableInfoSqlResult();
        List<Boolean> res = new ArrayList<>(tableInfoSqlResult.size());
        for (TableCompareRes<SqlCompareResult> tableCompareRes : tableInfoSqlResult) {
            // 初始化到DB、修改执行状态
            SyncEnvLog envLog = createSyncEnvLog(tableCompareRes);
            SyncEnvEventFactory.publishEvent(globalConfig, envLog, EvenType.SAVE_OR_UPDATE);
            // 准备执行
            CompareResult<SqlCompareResult> tableInfo = tableCompareRes.getTableInfo();
            List<CompareResult<SqlCompareResult>> compareResults = tableCompareRes.getCompareResults();
            List<BmScript> bmScripts = compareResults.stream().map(c -> c.getData().getScript()).collect(Collectors.toList());
            if (!bmScripts.isEmpty()) {
                String tableName = tableInfo.getTableName();
                // 把ddl执行 create table 等会存在多条SQL放一起,需要拆开 (只对pt中表级别的语句进行jdbc执行）
                List<CompareResult<SqlCompareResult>> modifyTableCRs = compareResults.stream().filter(c -> ModifyType.ModifyLevel.TABLE.equals(c.getModifyLevel())).collect(Collectors.toList());
                boolean success = true;// ddl语句可能为空，默认取true
                if (globalConfig.isSyncSandboxMock()) {
                    envLog.getSyncLog().add(LogUtils.formatStr("进入沙箱环境..."));
                }
                if (!modifyTableCRs.isEmpty()) {
                    List<String> stdDdl = modifyTableCRs.stream().map(c -> c.getData().getScript()).flatMap(d -> d.getSqls().stream()).filter(Objects::nonNull).collect(Collectors.toList());
                    long jdbcStart = System.currentTimeMillis();
                    try {
                        envLog.getSyncLog().add((LogUtils.formatStr( "开始执行JDBC DDL")));
                        List<String> allStdDdl = stdDdl.stream().flatMap(s -> {
                            List<SQLStatement> sqlStatements = SQLUtils.parseStatements(s, globalConfig.getDbType());
                            return sqlStatements.stream().map(SQLStatement::toString);
                        }).collect(Collectors.toList());
                        success = jdbcExecutor.executeBatch(allStdDdl);
                    } catch (ParserException e) {
                        log.error("druid SQL解析表{}的标准ddl失败，不进行后续的ptOsc执行", tableName, e);
                        LogUtils.logAndSet(globalConfig.getLogBizKey(), log, "druid SQL解析表%s的标准ddl失败，不进行后续的ptOsc执行！", tableName);
                        success = false;
                    } catch (SQLException e) {
                        log.error("执行表{}的标准ddl失败，不进行后续的ptOsc执行", tableName, e);
                        LogUtils.logAndSet(globalConfig.getLogBizKey(), log, "执行表%s的标准ddl失败，不进行后续的ptOsc执行！", tableName);
                        success = false;
                    } finally {
                        envLog.getSyncLog().add(LogUtils.formatStr("执行DDL，如下:"));
                        envLog.getSyncLog().add(LogUtils.formatStr( String.join(";\n", stdDdl)));
                        envLog.getSyncLog().add(LogUtils.formatStr("执行结果为：%s，耗时%dms",  (success ? "成功" : "失败"), System.currentTimeMillis() - jdbcStart));
                        envLog.getSyncLog().add(LogUtils.formatStr("完成执行JDBC DDL"));
                    }
                }
                List<String> ddls = null;
                if (globalConfig.getSyncToEnv().isPublishSync()) {
                    ddls = bmScripts.stream().flatMap(d -> d.getFullSqls().stream()).filter(Objects::nonNull).collect(Collectors.toList());
                } else {
                    ddls = bmScripts.stream().flatMap(d -> d.getSqls().stream()).filter(Objects::nonNull).collect(Collectors.toList());
                }
                List<String> ptOsc = bmScripts.stream().map(BmScript::getPtSql).filter(Objects::nonNull).collect(Collectors.toList());
                if (success && !ptOsc.isEmpty()) {
                    long ptStart = System.currentTimeMillis();
                    envLog.getSyncLog().add(LogUtils.formatStr( "开始执行PT OSC命令..."));
                    // pt-online-schema-change h=192.168.66.101,P=3306,D=ytest,t=data_million_small --user=root --ask-pass --alter "add comment varchar(50)" --charset=utf8 --execute
                    String ptDdl = String.join(",", ptOsc);
                    MysqlConnect mysqlConnect = globalConfig.getMysqlConnectByGlobalConfig();
                    final String syncLogPath = envLog.getSyncLogPath();
                    PtServerConfig ptServerConfig = globalConfig.getPtServerConfig();
                    final PtSsh ptSsh = ptServerConfig.getPtSsh();
                    PtOnlineSchemaChange ptOnlineSchemaChange = PtOnlineSchemaChange.builder()
                            .commandPath(ptServerConfig.getPtoscPath())
                            .host(mysqlConnect.getHost()).port(mysqlConnect.getPort())
                            .user(mysqlConnect.getUser()).password(mysqlConnect.getPassword())
                            .database(mysqlConnect.getDatabase()).table(tableName)
                            .initCustomProps()
                            .alter(ptDdl)
                            .execute(true)
                            .outputLogPath(syncLogPath)
                            .build();
                    final String script = ptOnlineSchemaChange.toString();
                    log.info("执行pt命令：{}", script);
                    envLog.getSyncLog().add(LogUtils.formatStr("原生DDL命令，如下：" ));
                    envLog.getSyncLog().add(LogUtils.formatStr( String.join("\n", ddls)));
                    envLog.getSyncLog().add(LogUtils.formatStr("执行ptosc命令，如下："));
                    envLog.getSyncLog().add(LogUtils.formatStr(ptOnlineSchemaChange.toString(true)));
                    Tuple2<Boolean, String> runScript = SshUtils.client().sshInfo(ptSsh).runScript(script);
                    success = runScript.getT1();
                    // 查看服务器日志
                    String catShell = "cat " + syncLogPath;
                    Tuple2<Boolean, String> catResult = SshUtils.client().sshInfo(ptSsh).runScript(catShell);
                    log.warn("执行查看日志命令{},status:{}", catShell, catResult.getT1());
                    envLog.getSyncLog().add(LogUtils.formatStr("执行过程，如下："));
                    envLog.getSyncLog().add(LogUtils.formatStr( catResult.getT2()));
                    // 耗时
                    envLog.getSyncLog().add(LogUtils.formatStr("执行结果为：%s，耗时%dms", success ? "成功" : "失败", System.currentTimeMillis() - ptStart));
                    envLog.getSyncLog().add(LogUtils.formatStr( "完成执行PT OSC命令"));
                }
                if (globalConfig.isSyncSandboxMock()) {
                    envLog.getSyncLog().add(LogUtils.formatStr("退出沙箱环境..."));
                }
                // 收集每次表执行情况
                res.add(success);
                envLog.setSyncScript(bmScripts);
                envLog.setSyncStatus(success ? SyncStatus.SYNC_DONE : SyncStatus.SYNC_ERROR);
                envLog.setSyncEndTime(LocalDateTime.now());
                SyncEnvEventFactory.publishEvent(globalConfig, envLog, EvenType.APPEND_UPDATE);
            }
        }
        // 没有一次为false则返回true
        return res.stream().noneMatch(Boolean.FALSE::equals);
    }

}
